package kafka

import (
	"fmt"
	"github.com/Shopify/sarama"
	"time"
)

var (
	cli     sarama.SyncProducer
	msgChan chan *sarama.ProducerMessage
)

func InitSyncProducer(addr []string, msg_size int32) (err error) {
	//1.生产者配置
	config := sarama.NewConfig()
	config.Producer.RequiredAcks = sarama.WaitForAll          //等待leader和follow确认
	config.Producer.Timeout = 30 * time.Second                //超时时间
	config.Producer.Partitioner = sarama.NewRandomPartitioner //分区
	config.Producer.Return.Successes = true                   //确认
	//2.连接kafka
	cli, err = sarama.NewSyncProducer(addr, config)
	//3.初始化信息通道
	msgChan = make(chan *sarama.ProducerMessage, msg_size)

	//4.开启发送goroutine
	go sendMsg()
	return
}

func sendMsg() {
	for {
		select {
		case msg := <-msgChan:
			fmt.Printf("msg.Value:%#v  msg.Topic:%s\n", msg.Value, msg.Topic)
			cli.SendMessage(msg)
		default:
			time.Sleep(time.Millisecond * 5)
		}
	}

}

func SendMessage(msg *sarama.ProducerMessage) {
	msgChan <- msg
}
